package rdbms.util;


import com.alibaba.druid.sql.parser.SQLParserUtils;
import com.alibaba.druid.sql.parser.SQLStatementParser;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import common.exception.DataExchangeException;
import common.util.Configuration;
import common.util.RetryableTasks;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.ImmutableTriple;
import org.apache.commons.lang3.tuple.Triple;
import rdbms.reader.Key;

import java.io.File;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.*;

@Slf4j
public final class Jdbcs
{


    private static final ThreadLocal<ExecutorService> rsExecutors = new ThreadLocal<ExecutorService>()
    {
        @Override
        protected ExecutorService initialValue()
        {
            return Executors.newFixedThreadPool(1, new ThreadFactoryBuilder()
                    .setNameFormat("rsExecutors-%d")
                    .setDaemon(true)
                    .build());
        }
    };

    private Jdbcs()
    {
    }

    public static String chooseJdbcUrl(
            final List<String> jdbcUrls, final String username,
            final String password, final List<String> preSql,
            final boolean checkSlave)
    {

        if (null == jdbcUrls || jdbcUrls.isEmpty())
        {
            throw DataExchangeException.asDataExchangeException(
                    DBUtilErrorCode.CONF_ERROR,
                    String.format("您的jdbcUrl的配置信息有错, 因为jdbcUrl[%s]不能为空. 请检查您的配置并作出修改.",
                            StringUtils.join(jdbcUrls, ",")));
        }

        try
        {
            return RetryableTasks.executeWithRetry(new Callable<String>()
            {

                public String call() throws Exception
                {
                    boolean connOK = false;
                    for (String url : jdbcUrls)
                    {
                        if (StringUtils.isNotBlank(url))
                        {
                            url = url.trim();
                            if (null != preSql && !preSql.isEmpty())
                            {
                                connOK = testConnWithoutRetry(
                                        url, username, password, preSql);
                            } else
                            {
                                connOK = testConnWithoutRetry(
                                        url, username, password, checkSlave);
                            }
                            if (connOK)
                            {
                                return url;
                            }
                        }
                    }
                    throw new Exception("DataX无法连接对应的数据库，可能原因是：1) 配置的ip/port/database/jdbc错误，无法连接。2) 配置的username/password错误，鉴权失败。请和DBA确认该数据库的连接信息是否正确。");
//                    throw new Exception(DBUtilErrorCode.JDBC_NULL.toString());
                }
            }, 7, 1000L, true);
            //warn: 7 means 2 minutes
        } catch (Exception e)
        {
            throw DataExchangeException.asDataExchangeException(
                    DBUtilErrorCode.CONN_DB_ERROR,
                    String.format("数据库连接失败. 因为根据您配置的连接信息,无法从:%s 中找到可连接的jdbcUrl. 请检查您的配置并作出修改.",
                            StringUtils.join(jdbcUrls, ",")), e);
        }
    }

    public static String chooseJdbcUrlWithoutRetry(
            final List<String> jdbcUrls, final String username,
            final String password, final List<String> preSql,
            final boolean checkSlave) throws DataExchangeException
    {

        if (null == jdbcUrls || jdbcUrls.isEmpty())
        {
            throw DataExchangeException.asDataExchangeException(
                    DBUtilErrorCode.CONF_ERROR,
                    String.format("您的jdbcUrl的配置信息有错, 因为jdbcUrl[%s]不能为空. 请检查您的配置并作出修改.",
                            StringUtils.join(jdbcUrls, ",")));
        }

        boolean connOK = false;
        for (String url : jdbcUrls)
        {
            if (StringUtils.isNotBlank(url))
            {
                url = url.trim();
                if (null != preSql && !preSql.isEmpty())
                {
                    connOK = testConnWithoutRetry(
                            url, username, password, preSql);
                } else
                {
                    try
                    {
                        connOK = testConnWithoutRetry(
                                url, username, password, checkSlave);
                    } catch (Exception e)
                    {
                        throw DataExchangeException.asDataExchangeException(
                                DBUtilErrorCode.CONN_DB_ERROR,
                                String.format("数据库连接失败. 因为根据您配置的连接信息,无法从:%s 中找到可连接的jdbcUrl. 请检查您的配置并作出修改.",
                                        StringUtils.join(jdbcUrls, ",")), e);
                    }
                }
                if (connOK)
                {
                    return url;
                }
            }
        }
        throw DataExchangeException.asDataExchangeException(
                DBUtilErrorCode.CONN_DB_ERROR,
                String.format("数据库连接失败. 因为根据您配置的连接信息,无法从:%s 中找到可连接的jdbcUrl. 请检查您的配置并作出修改.",
                        StringUtils.join(jdbcUrls, ",")));
    }

    /**
     * 检查slave的库中的数据是否已到凌晨00:00
     * 如果slave同步的数据还未到00:00返回false
     * 否则范围true
     *
     * @author ZiChi
     * @version 1.0 2014-12-01
     */
    private static boolean isSlaveBehind(Connection connection)
    {
        try
        {
            ResultSet resultSet = query(connection, "SHOW VARIABLES LIKE 'read_only'");
            if (Jdbcs.asyncResultSetNext(resultSet))
            {
                String $Value = resultSet.getString("Value");
                if ("ON".equalsIgnoreCase($Value))
                { //备库
                    ResultSet resultSet1 = query(connection, "SHOW SLAVE STATUS");
                    if (Jdbcs.asyncResultSetNext(resultSet1))
                    {
                        String $Slave_IO_Running = resultSet1.getString("Slave_IO_Running");
                        String $Slave_SQL_Running = resultSet1.getString("Slave_SQL_Running");
                        long $Seconds_Behind_Master = resultSet1.getLong("Seconds_Behind_Master");
                        if ("Yes".equalsIgnoreCase($Slave_IO_Running) && "Yes".equalsIgnoreCase($Slave_SQL_Running))
                        {
                            ResultSet resultSet2 = query(connection, "SELECT TIMESTAMPDIFF(SECOND, CURDATE(), NOW())");
                            Jdbcs.asyncResultSetNext(resultSet2);
                            long secondsOfDay = resultSet2.getLong(1);
                            return $Seconds_Behind_Master > secondsOfDay;
                        } else
                        {
                            return true;
                        }
                    } else
                    {
                        log.warn("SHOW SLAVE STATUS has no result");
                    }
                }
            } else
            {
                log.warn("SHOW VARIABLES like 'read_only' has no result");
            }
        } catch (Exception e)
        {
            log.warn("checkSlave failed, errorMessage:[{}].", e.getMessage());
        }
        return false;
    }


    public static boolean checkInsertPrivilege(String jdbcURL, String userName, String password, List<String> tableList)
    {
        Connection connection = connect(jdbcURL, userName, password);
        String insertTemplate = "insert into %s(select * from %s where 1 = 2)";

        boolean hasInsertPrivilege = true;
        Statement insertStmt = null;
        for (String tableName : tableList)
        {
            String checkInsertPrivilegeSql = String.format(insertTemplate, tableName, tableName);
            try
            {
                insertStmt = connection.createStatement();
                executeSqlWithoutResultSet(insertStmt, checkInsertPrivilegeSql);
            } catch (Exception e)
            {
                {
                    hasInsertPrivilege = false;
                    log.warn("User [" + userName + "] has no 'insert' privilege on tableName[" + tableName + "], errorMessage:[{}]", e.getMessage());
                }
            }
        }
        try
        {
            connection.close();
        } catch (SQLException e)
        {
            log.warn("connection close failed, " + e.getMessage());
        }
        return hasInsertPrivilege;
    }

    public static boolean checkDeletePrivilege(String jdbcURL, String userName, String password, List<String> tableList)
    {
        Connection connection = connect(jdbcURL, userName, password);
        String deleteTemplate = "delete from %s WHERE 1 = 2";

        boolean hasInsertPrivilege = true;
        Statement deleteStmt = null;
        for (String tableName : tableList)
        {
            String checkDeletePrivilegeSQL = String.format(deleteTemplate, tableName);
            try
            {
                deleteStmt = connection.createStatement();
                executeSqlWithoutResultSet(deleteStmt, checkDeletePrivilegeSQL);
            } catch (Exception e)
            {
                hasInsertPrivilege = false;
                log.warn("User [" + userName + "] has no 'delete' privilege on tableName[" + tableName + "], errorMessage:[{}]", e.getMessage());
            }
        }
        try
        {
            connection.close();
        } catch (SQLException e)
        {
            log.warn("connection close failed, " + e.getMessage());
        }
        return hasInsertPrivilege;
    }

    public static boolean needCheckDeletePrivilege(Configuration originalConfig)
    {
        List<String> allSqls = new ArrayList<String>();
        List<String> preSQLs = originalConfig.getList(Key.PRE_SQL, String.class);
        List<String> postSQLs = originalConfig.getList(Key.POST_SQL, String.class);
        if (preSQLs != null && !preSQLs.isEmpty())
        {
            allSqls.addAll(preSQLs);
        }
        if (postSQLs != null && !postSQLs.isEmpty())
        {
            allSqls.addAll(postSQLs);
        }
        for (String sql : allSqls)
        {
            if (StringUtils.isNotBlank(sql))
            {
                if (sql.trim().toUpperCase().startsWith("DELETE"))
                {
                    return true;
                }
            }
        }
        return false;
    }


    public static Connection getConnection(
            final String jdbcUrl, final String username, final String password)
    {
        return getConnection(jdbcUrl, username, password, String.valueOf(Constant.SOCKET_TIMEOUT_INSECOND * 1000));
    }

    /**
     * @param socketTimeout 设置socketTimeout，单位ms，String类型
     */
    public static Connection getConnection(
            final String jdbcUrl, final String username, final String password, final String socketTimeout)
    {

        try
        {
            return RetryableTasks.executeWithRetry(new Callable<Connection>()
            {
                public Connection call() throws Exception
                {
                    return Jdbcs.connect(jdbcUrl, username,
                            password, socketTimeout);
                }
            }, 9, 1000L, true);
        } catch (Exception e)
        {
            throw DataExchangeException.asDataExchangeException(
                    DBUtilErrorCode.CONN_DB_ERROR,
                    String.format("数据库连接失败. 因为根据您配置的连接信息:%s获取数据库连接失败. 请检查您的配置并作出修改.", jdbcUrl), e);
        }
    }

    /**
     * Get direct JDBC connection
     * <p/>
     * if connecting failed, try to connect for MAX_TRY_TIMES times
     * <p/>
     * NOTE: In DataX, we don't need connection pool in fact
     */
    public static Connection getConnectionWithoutRetry(
            final String jdbcUrl, final String username, final String password)
    {
        return getConnectionWithoutRetry(jdbcUrl, username,
                password, String.valueOf(Constant.SOCKET_TIMEOUT_INSECOND * 1000));
    }

    public static Connection getConnectionWithoutRetry(
            final String jdbcUrl, final String username, final String password, String socketTimeout)
    {
        return Jdbcs.connect(jdbcUrl, username,
                password, socketTimeout);
    }

    private static synchronized Connection connect(
            String url, String user, String pass)
    {
        return connect(url, user, pass, String.valueOf(Constant.SOCKET_TIMEOUT_INSECOND * 1000));
    }

    private static synchronized Connection connect(
            String url, String user, String pass, String socketTimeout)
    {


        Properties prop = new Properties();
        prop.put("user", user);
        prop.put("password", pass);


        return connect(url, prop);
    }

    private static synchronized Connection connect(
            String url, Properties properties)

    {
        try
        {

            DriverManager.setLoginTimeout(Constant.TIMEOUT_SECONDS);
            return DriverManager.getConnection(url, properties);
        } catch (Exception e)
        {
            throw RdbmsException.asConnException(e, properties.getProperty("user"), null);
        }
    }

    /**
     * a wrapped method to execute select-like sql statement .
     *
     * @param connection Database connection .
     * @param sql        sql statement to be executed
     * @return a {@link ResultSet}
     * @throws SQLException if occurs SQLException.
     */
    public static ResultSet query(Connection connection, String sql, int fetchSize)
            throws SQLException
    {
        // 默认3600 s 的query Timeout
        return query(connection, sql, fetchSize, Constant.SOCKET_TIMEOUT_INSECOND);
    }

    /**
     * a wrapped method to execute select-like sql statement .
     *
     * @param connection   Database connection .
     * @param sql          sql statement to be executed
     * @param fetchSize
     * @param queryTimeout unit:second
     * @return
     * @throws SQLException
     */
    public static ResultSet query(Connection connection, String sql, int fetchSize, int queryTimeout)
            throws SQLException
    {
        // make sure autocommit is off
        connection.setAutoCommit(false);
        Statement statement = connection.createStatement(ResultSet.TYPE_FORWARD_ONLY,
                ResultSet.CONCUR_READ_ONLY);
        statement.setFetchSize(fetchSize);
        statement.setQueryTimeout(queryTimeout);
        return query(statement, sql);
    }

    /**
     * a wrapped method to execute select-like sql statement .
     *
     * @param statement {@link Statement}
     * @param sql       sql statement to be executed
     * @return a {@link ResultSet}
     * @throws SQLException if occurs SQLException.
     */
    public static ResultSet query(Statement statement, String sql)
            throws SQLException
    {
        return statement.executeQuery(sql);
    }

    public static void executeSqlWithoutResultSet(Statement statement, String sql)
            throws SQLException
    {
        statement.execute(sql);
    }

    /**
     * Close {@link ResultSet}, {@link Statement} referenced by this
     * {@link ResultSet}
     *
     * @param resultSet {@link ResultSet} to be closed
     * @throws IllegalArgumentException
     */
    public static void closeResultSet(ResultSet resultSet)
    {
        try
        {
            if (null != resultSet)
            {
                Statement statement = resultSet.getStatement();
                if (null != statement)
                {
                    statement.close();
                    statement = null;
                }
                resultSet.close();
            }
            resultSet = null;
        } catch (SQLException e)
        {
            throw new IllegalStateException(e);
        }
    }

    public static void close(ResultSet resultSet, Statement statement,
                             Connection connection)
    {
        if (null != resultSet)
        {
            try
            {
                resultSet.close();
            } catch (SQLException unused)
            {
            }
        }

        if (null != statement)
        {
            try
            {
                statement.close();
            } catch (SQLException unused)
            {
            }
        }

        if (null != connection)
        {
            try
            {
                connection.close();
            } catch (SQLException unused)
            {
            }
        }
    }

    public static void close(Statement statement, Connection connection)
    {
        close(null, statement, connection);
    }

    public static List<String> getTableColumns(
            String jdbcUrl, String user, String pass, String tableName)
    {
        Connection connection = getConnection(jdbcUrl, user, pass);
        return getTableColumnsByConn(connection, tableName, "jdbcUrl:" + jdbcUrl);
    }

    public static List<String> getTableColumnsByConn(Connection connection, String tableName, String basicMsg)
    {
        List<String> columnNameList = new ArrayList<String>();
        Statement statement = null;
        ResultSet resultSet = null;
        String queryColumnSql = null;
        try
        {
            statement = connection.createStatement();
            queryColumnSql = String.format("select * from %s where 1=2",
                    tableName);
            resultSet = statement.executeQuery(queryColumnSql);
            ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
            for (int i = 0, columnCount = resultSetMetaData.getColumnCount(); i < columnCount; i++)
            {
                columnNameList.add(resultSetMetaData.getColumnName(i + 1));
            }

        } catch (SQLException e)
        {
            throw RdbmsException.asQueryException(e, queryColumnSql, tableName, null);
        } finally
        {
            Jdbcs.close(resultSet, statement, connection);
        }

        return columnNameList;
    }


    /**
     * @return Left:ColumnName Middle:ColumnType Right:ColumnTypeName
     */
    public static Triple<List<String>, List<Integer>, List<String>> getColumnMetaData(
            Connection connection, String tableName, String column)
    {
        Statement statement = null;
        ResultSet resultSet = null;

        Triple<List<String>, List<Integer>, List<String>> columnMetaData = new ImmutableTriple<List<String>, List<Integer>, List<String>>(
                new ArrayList<String>(), new ArrayList<Integer>(),
                new ArrayList<String>());
        try
        {
            statement = connection.createStatement();
            String queryColumnSql = "select " + column + " from " + tableName
                    + " where 1=2";

            resultSet = statement.executeQuery(queryColumnSql);
            ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
            for (int i = 0, columnCount = resultSetMetaData.getColumnCount(); i < columnCount; i++)
            {

                columnMetaData.getLeft().add(resultSetMetaData.getColumnName(i + 1));
                columnMetaData.getMiddle().add(resultSetMetaData.getColumnType(i + 1));
                columnMetaData.getRight().add(
                        resultSetMetaData.getColumnTypeName(i + 1));
            }
            return columnMetaData;

        } catch (SQLException e)
        {
            throw DataExchangeException
                    .asDataExchangeException(DBUtilErrorCode.GET_COLUMN_INFO_FAILED,
                            String.format("获取表:%s 的字段的元信息时失败. 请联系 DBA 核查该库、表信息.", tableName), e);
        } finally
        {
            Jdbcs.close(resultSet, statement, null);
        }
    }

    public static boolean testConnWithoutRetry(
            String url, String user, String pass, boolean checkSlave)
    {
        Connection connection = null;

        try
        {
            connection = connect(url, user, pass);
            if (connection != null)
            {
                if (checkSlave)
                {
                    //dataBaseType.MySql
                    boolean connOk = !isSlaveBehind(connection);
                    return connOk;
                } else
                {
                    return true;
                }
            }
        } catch (Exception e)
        {
            log.warn("test connection of [{}] failed, for {}.", url,
                    e.getMessage());
        } finally
        {
            Jdbcs.close(null, connection);
        }
        return false;
    }

    public static boolean testConnWithoutRetry(
            String url, String user, String pass, List<String> preSql)
    {
        Connection connection = null;
        try
        {
            connection = connect(url, user, pass);
            if (null != connection)
            {
                for (String pre : preSql)
                {
                    if (doPreCheck(connection, pre) == false)
                    {
                        log.warn("doPreCheck failed.");
                        return false;
                    }
                }
                return true;
            }
        } catch (Exception e)
        {
            log.warn("test connection of [{}] failed, for {}.", url,
                    e.getMessage());
        } finally
        {
            Jdbcs.close(null, connection);
        }

        return false;
    }


    public static ResultSet query(Connection connection, String sql)
            throws SQLException
    {
        Statement statement = connection.createStatement(ResultSet.TYPE_FORWARD_ONLY,
                ResultSet.CONCUR_READ_ONLY);
        //默认3600 seconds
        statement.setQueryTimeout(Constant.SOCKET_TIMEOUT_INSECOND);
        return query(statement, sql);
    }

    private static boolean doPreCheck(Connection connection, String pre)
    {
        ResultSet resultSet = null;
        try
        {
            resultSet = query(connection, pre);

            int checkResult = -1;
            if (Jdbcs.asyncResultSetNext(resultSet))
            {
                checkResult = resultSet.getInt(1);
                if (Jdbcs.asyncResultSetNext(resultSet))
                {
                    log.warn(
                            "pre check failed. It should return one result:0, pre:[{}].",
                            pre);
                    return false;
                }

            }

            if (0 == checkResult)
            {
                return true;
            }

            log.warn(
                    "pre check failed. It should return one result:0, pre:[{}].",
                    pre);
        } catch (Exception e)
        {
            log.warn("pre check failed. pre:[{}], errorMessage:[{}].", pre,
                    e.getMessage());
        } finally
        {
            Jdbcs.closeResultSet(resultSet);
        }
        return false;
    }

    // warn:until now, only oracle need to handle session config.
    public static void dealWithSessionConfig(Connection connection,
                                             Configuration configuration, String message)
    {
        List<String> sessionConfig = null;
        sessionConfig = configuration.getList(Key.SESSION,
                new ArrayList<String>(), String.class);
        Jdbcs.doDealWithSessionConfig(connection, sessionConfig, message);

    }

    private static void doDealWithSessionConfig(Connection connection,
                                                List<String> sessions, String message)
    {
        if (null == sessions || sessions.isEmpty())
        {
            return;
        }

        Statement stmt;
        try
        {
            stmt = connection.createStatement();
        } catch (SQLException e)
        {
            throw DataExchangeException
                    .asDataExchangeException(DBUtilErrorCode.SET_SESSION_ERROR, String
                                    .format("session配置有误. 因为根据您的配置执行 session 设置失败. 上下文信息是:[%s]. 请检查您的配置并作出修改.", message),
                            e);
        }

        for (String sessionSql : sessions)
        {
            log.info("execute sql:[{}]", sessionSql);
            try
            {
                Jdbcs.executeSqlWithoutResultSet(stmt, sessionSql);
            } catch (SQLException e)
            {
                throw DataExchangeException.asDataExchangeException(
                        DBUtilErrorCode.SET_SESSION_ERROR, String.format(
                                "session配置有误. 因为根据您的配置执行 session 设置失败. 上下文信息是:[%s]. 请检查您的配置并作出修改.", message), e);
            }
        }
        Jdbcs.close(stmt, null);
    }

    public static void sqlValid(String sql)
    {
        SQLStatementParser statementParser = SQLParserUtils.createSQLStatementParser(sql, null);
        statementParser.parseStatementList();
    }

    /**
     * 异步获取resultSet的next(),注意，千万不能应用在数据的读取中。只能用在meta的获取
     *
     * @param resultSet
     * @return
     */
    public static boolean asyncResultSetNext(final ResultSet resultSet)
    {
        return asyncResultSetNext(resultSet, 3600);
    }

    public static boolean asyncResultSetNext(final ResultSet resultSet, int timeout)
    {
        Future<Boolean> future = rsExecutors.get().submit(new Callable<Boolean>()
        {
            public Boolean call() throws Exception
            {
                return resultSet.next();
            }
        });
        try
        {
            return future.get(timeout, TimeUnit.SECONDS);
        } catch (Exception e)
        {
            throw DataExchangeException.asDataExchangeException(
                    DBUtilErrorCode.RS_ASYNC_ERROR, "异步获取ResultSet失败", e);
        }
    }

    public static void loadDriverClass(String pluginType, String pluginName)
    {
        try
        {
            String pluginJsonPath = StringUtils.join(
                    new String[]{System.getProperty("datax.home"), "plugin",
                            pluginType,
                            String.format("%s%s", pluginName, pluginType),
                            "plugin.json"}, File.separator);
            Configuration configuration = Configuration.from(new File(
                    pluginJsonPath));
            List<String> drivers = configuration.getList("drivers",
                    String.class);
            for (String driver : drivers)
            {
                Class.forName(driver);
            }
        } catch (ClassNotFoundException e)
        {
            throw DataExchangeException.asDataExchangeException(DBUtilErrorCode.CONF_ERROR,
                    "数据库驱动加载错误, 请确认libs目录有驱动jar包且plugin.json中drivers配置驱动类正确!",
                    e);
        }
    }
}
